package com.eatme.web.app.coupon.distribution.service.impl;

import com.alibaba.fastjson2.JSON;
import com.eatme.web.app.coupon.common.constant.Constant;
import com.eatme.web.app.coupon.distribution.constant.CouponStatus;
import com.eatme.web.app.coupon.distribution.vo.CouponKafkaMessage;
import com.eatme.web.app.coupon.distribution.dao.CouponDao;
import com.eatme.web.app.coupon.distribution.entity.Coupon;
import com.eatme.web.app.coupon.distribution.service.IKafaService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Optional;

/**
 * Kafka 相关的服务接口实现
 * 核心思想: 是将 Cache中的Coupon 的状态变化同步到 DB 中
 */
@Service
public class KafkaServiceImpl implements IKafaService {
    private static final Logger LOG = LoggerFactory.getLogger(KafkaServiceImpl.class);

    private final CouponDao couponDao;

    public KafkaServiceImpl(CouponDao couponDao) {
        this.couponDao = couponDao;
    }


    /**
     * 消费优惠券 Kafka 消息
     * @param record {@link ConsumerRecord}
     */
    @Override
    @KafkaListener(topics = {Constant.TOPIC},groupId = "app-coupon-1")
    public void consumeCouponKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            CouponKafkaMessage couponInfo = JSON.parseObject(
                    message.toString(),
                    CouponKafkaMessage.class);
            LOG.info("Receive CouponKafkaMessage: {}",message.toString());
            CouponStatus couponStatus = CouponStatus.of(couponInfo.getStatus());

            switch (couponStatus) {
                case USABLE:
                    break;
                case USED: precessUsedCoupon(couponInfo,couponStatus);
                    break;
                case EXPIRED: precessExpiredCoupon(couponInfo,couponStatus);
                    break;


            }

        }
    }

    /**
     * 处理已使用的用户优惠券
     * @param kafkaMessage
     * @param status
     */
    private void precessUsedCoupon(CouponKafkaMessage kafkaMessage,
                                        CouponStatus status) {
        // TODO 给用户发送短信提醒
        precessCouponsByStatus(kafkaMessage,status);

    }
    /**
     * 处理过期的用户优惠券
     * @param kafkaMessage
     * @param status
     */
    private void precessExpiredCoupon(CouponKafkaMessage kafkaMessage,
                                   CouponStatus status) {
        // TODO 给用户发送推送
        precessCouponsByStatus(kafkaMessage,status);

    }

    /**
     * 根据状态处理优惠券信息
     * @param kafkaMessage
     * @param status
     */
    private void precessCouponsByStatus(CouponKafkaMessage kafkaMessage,
                                        CouponStatus status) {
        List<Coupon> coupons = couponDao.findAllById(kafkaMessage.getIds());

        if (CollectionUtils.isEmpty(coupons)
                || coupons.size() != kafkaMessage.getIds().size()) {
            LOG.error("Cant NOt Find Right Coupon Info: {}",JSON.toJSONString(kafkaMessage));
            // TODO 发送邮件
            return;
        }

        coupons.forEach(coupon -> coupon.setStatus(status));
        LOG.info("CouponKafkaMessage Op Coupon Count: {}",couponDao.saveAll(coupons).size());
    }
}
